package org.example.flinktest.apidemo.sinktest;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig.Builder;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.example.flinktest.bean.SensorReading;
import org.example.flinktest.operators.MakeSensorMapFunction;

/**
 * @author shihb
 * @date 2019/12/11 10:42
 */
public class RedisSinkDemo {

  public static void main(String[] args) throws Exception {
    //环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);

    //source
    DataStream<String> input = env.readTextFile(
        "D:\\SHBData\\IDEAProjects\\fink-parent\\flink-quickstart-java\\src\\main\\resources\\sensor.txt");



    //transform操作
    DataStream<SensorReading> dataStream = input.map(new MakeSensorMapFunction());


    //sink
    FlinkJedisPoolConfig conf = new Builder()
        .setHost("localhost")
        .setPort(7001)
        .setPassword("NL128redis")
        .build();
    dataStream.addSink(new RedisSink<>(conf, new MyRedisMapper()));


    env.execute("redis sink test");
  }

}
/**
 * 我们设计该数据的在redis的存储结构为hash，redis key为sensor_temperature
 * hash 键值对为 id:temperature
 */
class MyRedisMapper implements RedisMapper<SensorReading> {

  //定义要使用的redis命令
  @Override
  public RedisCommandDescription getCommandDescription() {

    return new RedisCommandDescription(RedisCommand.HSET,"sensor_temperature");
  }

  //定义键值对的key
  @Override
  public String getKeyFromData(SensorReading input) {
    return input.getId();
  }

  //定义键值对的key的value
  @Override
  public String getValueFromData(SensorReading input) {
    return input.getTemperature().toString();
  }
}
